/*
 ============================================================================
 Name        : ThreadsPPD.c
 Author      : SOogle
 Version     : 1.0
 Copyright   : All rights reserved
 Description : FUSELAGE - FUSE FAT32 FILE SYSTEM
 ============================================================================
 */

#include "../../headers/GlobalHeader.h"

extern int32_t cantidadClientesPFS;
extern struct_config configuracion;
extern pthread_mutex_t mutexCola;
int32_t contador_msync;
extern t_log *logger;
extern int32_t filesize;
extern char* handlerArchivoMapeado;
int32_t sd;
extern int32_t sizeCola;
extern PPD_CHS posicionActualCabezal;
extern NodoPFS *listaPFS;
extern int32_t algoritmoUtilizado;
extern int32_t cantidadClientesPFS;

void thread_lector_pedidos_raid (void *parametro)
{
	struct_thread_lector struct_thread = *(struct_thread_lector *) parametro;
	int32_t socket_id = struct_thread.socket_id;
	int32_t rc;
	nipc paquete;

	while(1)
	{
		rc = recv(socket_id,&paquete,sizeof(nipc),MSG_WAITALL);
		if (rc < 0)
		{
			if (configuracion.LogActivado==1)
				log_error(logger, "THREAD_LECTOR_PEDIDOS_RAID", "Error al recibir pedido de RAID");
			cerrarArchivo();
			exit(1);
		}

		pthread_mutex_lock (&mutexCola);
		agregarPedido_cola(paquete.payload);
		pthread_mutex_unlock (&mutexCola);
		PPD_CHS estructura;
		//printf("me llego %i de %c ;;",paquete.payload.direccionSector,paquete.payload.lectura_o_escritura);
		estructura = ppd_LBAToCHS(paquete.payload.direccionSector);
		if (configuracion.LogActivado==1)
			log_info(logger, "THREAD_LECTOR_PEDIDOS_RAID", "Se agrego el pedido de %c sobre el sector %i:%i correctamente\n", paquete.payload.lectura_o_escritura, estructura.cylinder, estructura.sector);
	}
}

void thread_msync ()
{

	typedef struct timespec
	{
		time_t   tv_sec;   /* seconds */
		long     tv_nsec;  /* nanoseconds */
	}timespec_t;

	int32_t valorRetorno;
	timespec_t tim;
	tim.tv_nsec = 500000000;

	while(TRUE)
	{
		if(contador_msync > 30)
		{
			if((valorRetorno=msync(handlerArchivoMapeado, filesize*TO_BYTES, MS_SYNC)) == -1 ){
				perror("msync erroneo1");
				cerrarArchivo();
				exit(1);
			}
			contador_msync = 0;
		}
		else
		{
			nanosleep(&tim,NULL);
			if((valorRetorno=msync(handlerArchivoMapeado, filesize*TO_BYTES, MS_SYNC)) == -1 ){
				perror("msync erroneo2");
				cerrarArchivo();
				exit(TRUE);
			}
		}
	}
}

void sigchld_handler(int32_t signal)
{
	while(waitpid(-1, 0, WNOHANG));
}


void thread_lector_consola (void *parametro)
{
	int32_t resultado;
	cabecera_paquete cabeceraRecibo;
	protocoloINFO_respuesta *respuestaINFO;
	pedido_clean pedidoClean;
	pedido_t1 pedidoTrace1;
	pedido_t2 pedidoTrace2;
	pedido_t3 pedidoTrace3;
	pedido_t4 pedidoTrace4;
	pedido_t5 pedidoTrace5;
	t_stream *streamEnvio;
	t_pedido pedidoAPlanificar;
	int32_t rc, contador, cantidadSectores, current, i;
	int32_t *sectoresAPlanificar;
	struct sockaddr_un serveraddr;
	char sectoresPorPista[10];
	char cabezalesPorCilindro[2];
	char cilindros[10];
	sprintf(sectoresPorPista,"%d",configuracion.Sectors);
	sprintf(cabezalesPorCilindro, "%d", configuracion.Heads);
	sprintf(cilindros, "%d", configuracion.Cylinders);

	char* const args[] = {configuracion.SocketUNIX,cabezalesPorCilindro,sectoresPorPista,cilindros, NULL};

	PPD_CHS sectorPlanifico, sectorInicial, sectorFinal;

	int32_t child_pid;

	switch (child_pid = fork())
	{
		case -1 :

			perror("fork");
			break;

		case 0 : /*LLamada al proceso Consola PPD*/
			resultado = execv (configuracion.PathConsola, args);
			if (resultado == -1)
				perror("execv");
			cerrarArchivo();
			break;

		default : /*Codigo del padre, aca seguiria*/

			unlink(configuracion.SocketUNIX);
			sd = socket(AF_UNIX, SOCK_STREAM, 0);
			if (sd < 0)
			{
				perror("socket() failed");
				cerrarArchivo();
				break;
			}

			memset(&serveraddr, 0, sizeof(serveraddr));
			serveraddr.sun_family = AF_UNIX;
			strcpy(serveraddr.sun_path, configuracion.SocketUNIX);

			rc = bind(sd, (struct sockaddr *)&serveraddr, SUN_LEN(&serveraddr));
			if (rc < 0)
			{
				perror("bind() failed");
				cerrarArchivo();
				break;
			}

			rc = listen(sd, 10);
			if (rc< 0)
			{
				perror("listen() failed");
				cerrarArchivo();
				break;
			}

			sd = accept(sd, NULL, NULL);
			if (sd < 0)
			{
				perror("accept() failed");
				cerrarArchivo();
				break;
			}

			while(1)
			{

				rc = recv(sd, &cabeceraRecibo, sizeof(cabeceraRecibo), MSG_WAITALL);
				if (rc < 0)
				{
					perror("recv() failed");
					cerrarArchivo();
					break;
				}

				switch(cabeceraRecibo.type)
				{
					case TEMPO:

						rc = recv(sd, &pedidoTrace1, cabeceraRecibo.payloadlength, MSG_WAITALL);

						//UTILIZO EL MISMO STRUCT DEL PEDIDO TRACE 1, YA QUE ES UN COMANDO Y UN DIGITO
						configuracion.TiempoLectura = 1000000*pedidoTrace1.s1;

						break;

					case SIZEPEDIDOS:

						//UTILIZO EL MISMO STRUCT QUE EL INFO YA QUE ES EL MISMO FORMATO DE COMANDO

						respuestaINFO = malloc(sizeof(protocoloINFO_respuesta));
						respuestaINFO->cabecera.type = SIZEPEDIDOS;
						respuestaINFO->cabecera.payloadlength = sizeof(int32_t);
						respuestaINFO->posicionActual = sizeCola;

						streamEnvio = RespuestaINFO_serializer(respuestaINFO);

						rc = send(sd, streamEnvio->data, streamEnvio->length, 0);

						free(respuestaINFO);
						free(streamEnvio);

						break;

					case INFOCONSOLA:

						respuestaINFO = malloc(sizeof(protocoloINFO_respuesta));
						respuestaINFO->cabecera.type = INFOCONSOLA;
						respuestaINFO->cabecera.payloadlength = sizeof(int32_t);
						respuestaINFO->posicionActual = ppd_CHStoLBA(posicionActualCabezal);

						streamEnvio = RespuestaINFO_serializer(respuestaINFO);

						rc = send(sd, streamEnvio->data, streamEnvio->length, 0);

						free(respuestaINFO);
						free(streamEnvio);
						break;

					case CLEAN:

						rc = recv(sd, &pedidoClean, cabeceraRecibo.payloadlength, MSG_WAITALL);

						//armo tantos pedidos como diferencia haya entre los sectores que recibo
						sectorInicial = ppd_LBAToCHS(pedidoClean.sectorInicial);
						sectorFinal = ppd_LBAToCHS(pedidoClean.sectorFinal);

						cantidadSectores = diferenciaEntreSectores(sectorInicial, sectorFinal);

						for (i = 0; i <= cantidadSectores; i++)
						{
							pedidoAPlanificar = armoPedidoAPlanificar(pedidoClean.sectorInicial+i, 'E');
							PPD_CHS estructura;
							estructura = ppd_LBAToCHS(pedidoAPlanificar.direccionSector);

							pthread_mutex_lock (&mutexCola);
							agregarPedido_cola(pedidoAPlanificar);
							if (configuracion.LogActivado==1)
								log_info(logger, "PPD - THREAD LECTOR PEDIDOS CONSOLA", "Se agrego un pedido de %c sobre el sector %d:%d a la cola \n", pedidoAPlanificar.lectura_o_escritura, estructura.cylinder, estructura.sector);
							pthread_mutex_unlock (&mutexCola);
						}
						break;

					case TRACE1:

						rc = recv(sd, &pedidoTrace1, cabeceraRecibo.payloadlength, MSG_WAITALL);

						sectorPlanifico = ppd_LBAToCHS(pedidoTrace1.s1);

						pedidoAPlanificar = armoPedidoAPlanificar(pedidoTrace1.s1, 'L');
						PPD_CHS estructura;
						estructura = ppd_LBAToCHS(pedidoAPlanificar.direccionSector);

						pthread_mutex_lock (&mutexCola);
						agregarPedido_cola(pedidoAPlanificar);
						if (configuracion.LogActivado==1)
							log_info(logger, "PPD - THREAD LECTOR PEDIDOS CONSOLA", "Se agrego un pedido de %c sobre el sector %d:%d a la cola \n", pedidoAPlanificar.lectura_o_escritura, estructura.cylinder, estructura.sector);
						pthread_mutex_unlock (&mutexCola);

						break;

					case TRACE2:

						rc = recv(sd, &pedidoTrace2, cabeceraRecibo.payloadlength, MSG_WAITALL);

						contador = 0;
						current = 0;
						sectoresAPlanificar = malloc(sizeof(int32_t)*2);
						memcpy(sectoresAPlanificar+current,&pedidoTrace2.s1, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace2.s2, sizeof(int32_t));


						while (contador < 2)
						{
							sectorPlanifico = ppd_LBAToCHS(sectoresAPlanificar[contador]);

							pedidoAPlanificar = armoPedidoAPlanificar(ppd_CHStoLBA(sectorPlanifico),'L');
							PPD_CHS estructura;
							estructura = ppd_LBAToCHS(pedidoAPlanificar.direccionSector);

							pthread_mutex_lock (&mutexCola);
							agregarPedido_cola(pedidoAPlanificar);
							if (configuracion.LogActivado==1)
								log_info(logger, "PPD - THREAD LECTOR PEDIDOS CONSOLA", "Se agrego un pedido de %c sobre el sector %d:%d a la cola \n", pedidoAPlanificar.lectura_o_escritura, estructura.cylinder, estructura.sector);
							pthread_mutex_unlock (&mutexCola);

							contador++;
						}

						free(sectoresAPlanificar);
						break;

					case TRACE3:

						rc = recv(sd, &pedidoTrace3, cabeceraRecibo.payloadlength, MSG_WAITALL);

						contador = 0;
						current = 0;
						sectoresAPlanificar = malloc(sizeof(int32_t)*3);
						memcpy(sectoresAPlanificar+current,&pedidoTrace3.s1, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace3.s2, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace3.s3, sizeof(int32_t));

						while(contador<3)
						{
							sectorPlanifico = ppd_LBAToCHS(sectoresAPlanificar[contador]);

							pedidoAPlanificar = armoPedidoAPlanificar(ppd_CHStoLBA(sectorPlanifico),'L');
							PPD_CHS estructura;
							estructura = ppd_LBAToCHS(pedidoAPlanificar.direccionSector);

							pthread_mutex_lock (&mutexCola);
							agregarPedido_cola(pedidoAPlanificar);
							if (configuracion.LogActivado==1)
								log_info(logger, "PPD - THREAD LECTOR PEDIDOS CONSOLA", "Se agrego un pedido de %c sobre el sector %d:%d a la cola \n", pedidoAPlanificar.lectura_o_escritura, estructura.cylinder, estructura.sector);
							pthread_mutex_unlock (&mutexCola);


							contador++;
						}

						free(sectoresAPlanificar);
						break;

					case TRACE4:

						rc = recv(sd, &pedidoTrace4, cabeceraRecibo.payloadlength, MSG_WAITALL);

						contador = 0;
						current = 0;
						sectoresAPlanificar = malloc(sizeof(int32_t)*4);
						memcpy(sectoresAPlanificar+current,&pedidoTrace4.s1, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace4.s2, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace4.s3, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace4.s4, sizeof(int32_t));

						while(contador < 4)
						{
							sectorPlanifico = ppd_LBAToCHS(sectoresAPlanificar[contador]);

							pedidoAPlanificar = armoPedidoAPlanificar(ppd_CHStoLBA(sectorPlanifico),'L');
							PPD_CHS estructura;
							estructura = ppd_LBAToCHS(pedidoAPlanificar.direccionSector);

							pthread_mutex_lock (&mutexCola);
							agregarPedido_cola(pedidoAPlanificar);
							if (configuracion.LogActivado==1)
								log_info(logger, "PPD - THREAD LECTOR PEDIDOS CONSOLA", "Se agrego un pedido de %c sobre el sector %d:%d a la cola \n", pedidoAPlanificar.lectura_o_escritura, estructura.cylinder, estructura.sector);
							pthread_mutex_unlock (&mutexCola);

							contador++;
						}

						free(sectoresAPlanificar);
						break;

					case TRACE5:

						rc = recv(sd, &pedidoTrace5, cabeceraRecibo.payloadlength, MSG_WAITALL);

						contador = 0;
						current = 0;
						sectoresAPlanificar = malloc(sizeof(int32_t)*5);
						memcpy(sectoresAPlanificar+current,&pedidoTrace5.s1, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace5.s2, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace5.s3, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace5.s4, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace5.s5, sizeof(int32_t));

						while(contador < 5)
						{
							sectorPlanifico = ppd_LBAToCHS(sectoresAPlanificar[contador]);

							pedidoAPlanificar = armoPedidoAPlanificar(ppd_CHStoLBA(sectorPlanifico),'L');
							PPD_CHS estructura;
							estructura = ppd_LBAToCHS(pedidoAPlanificar.direccionSector);

							pthread_mutex_lock (&mutexCola);
							agregarPedido_cola(pedidoAPlanificar);
							if (configuracion.LogActivado==1)
								log_info(logger, "PPD - THREAD LECTOR PEDIDOS CONSOLA", "Se agrego un pedido de %c sobre el sector %d:%d a la cola \n", pedidoAPlanificar.lectura_o_escritura, estructura.cylinder, estructura.sector);
							pthread_mutex_unlock (&mutexCola);

							contador++;
						}

						free(sectoresAPlanificar);
						break;
				}

				if (rc < 0)
				{
					perror("send() failed");
					cerrarArchivo();
					break;
				}
				signal(SIGCHLD, sigchld_handler);
			}
			unlink(configuracion.SocketUNIX);
			close(sd);
			break;
	}
}

void thread_lector_pedidos_pfs (void *parametro)
{
	nipc paquete_recibo, paquete_mando;
	int32_t addrlen = sizeof(struct sockaddr_in);
	int32_t socket_server_pfs = socket(AF_INET,SOCK_STREAM,0);
	int32_t cliente_pfs, rc;
	fd_set sockets_pfs, sockets_select;

	int32_t descriptor_maximo = 0;

	PPD_CHS estructura;

	//DIRECCION LOCAL PFS
	struct sockaddr_in *direccion_local_pfs = malloc(addrlen);

	//DIRECCION REMOTA - A LA QUE ME QUIERO CONECTAR -
	struct sockaddr_in *direccion_remota_pfs = malloc(addrlen);

	direccion_local_pfs->sin_family = AF_INET;
	direccion_local_pfs->sin_addr.s_addr = inet_addr("127.0.0.1");

	direccion_local_pfs->sin_port = htons(configuracion.PuertoEscuchaPpd);
	bind(socket_server_pfs,(struct sockaddr *)direccion_local_pfs, addrlen);

	if (configuracion.LogActivado==1)
	{
		log_info(logger, "PPD - THREAD LECTOR PEDIDOS PFS", "El PPD se encuentra a la espera de conexiones entrantes \n");
	}

	rc = listen(socket_server_pfs,CANT_CLIENTES_PFS_MAX);
	if (rc < 0)
	{
		if (configuracion.LogActivado==1)
			log_error(logger, "PPD - THREAD LECTOR PEDIDOS PFS", "El PPD ha esperado demasiado por conexiones entrantes");
		cerrarArchivo();
		perror("listen() failed");
	}

	FD_ZERO (&sockets_pfs);
	FD_SET(socket_server_pfs,&sockets_pfs);
	descriptor_maximo = socket_server_pfs;

	while(1)
	{
		sockets_select = sockets_pfs;

		select(descriptor_maximo+1,&sockets_select,NULL,NULL,NULL);

		if(FD_ISSET(socket_server_pfs,&sockets_select))
		{
			rc = accept(socket_server_pfs,(struct sockaddr *)direccion_remota_pfs,(void *)&addrlen);
			cliente_pfs = rc;
			if (configuracion.LogActivado==1)
				log_info(logger, "PPD - THREAD LECTOR PEDIDOS PFS", "Se acepto la conexion del CLIENTE PFS %i", cliente_pfs);

			//INSERTO EN LA LISTA DE CLIENTES DEL PPD
			insertarEnListaClientes(cliente_pfs);
			FD_SET(cliente_pfs,&sockets_pfs);
			if(cliente_pfs > descriptor_maximo)
				descriptor_maximo = cliente_pfs;

		}

		NodoPFS *auxPFS=NULL;

		auxPFS=listaPFS;

		while(auxPFS!=NULL){
			if(FD_ISSET (auxPFS->cliente_pfs,&sockets_select))
			{
				rc = recv(auxPFS->cliente_pfs,&paquete_recibo,sizeof(nipc),0);
				switch(rc)
				{
					case 0: //EL CLIENTE CERRO LA CONEXION
						borrarDeListaClientes(auxPFS->cliente_pfs);
						close(auxPFS->cliente_pfs);
						if (configuracion.LogActivado==1)
							log_error(logger, "PPD - THREAD LECTOR PEDIDOS PFS", "El cliente %i cerro la conexion", auxPFS->cliente_pfs);
						break;
					case -1: //ERROR EN EL RECV
						if (configuracion.LogActivado==1)
							log_error(logger, "PPD - THREAD LECTOR PEDIDOS PFS", "Error al recibir el paquete, se cierra la conexion del cliente %i", auxPFS->cliente_pfs);
						close(auxPFS->cliente_pfs);
						break;
					default: //SE RECIBIO CORRECTAMENTE
						switch(paquete_recibo.type)
						{
							case HANDSHAKE:
								if(paquete_recibo.payloadlength == 0)
								{
									paquete_mando.type = HANDSHAKE;
									paquete_mando.payloadlength = 0;
									paquete_mando.payload.conexion_pfs_id = auxPFS->cliente_pfs;
									if (configuracion.LogActivado==1)
										log_info(logger, "PPD - THREAD LECTOR PEDIDOS PFS", "Se recibio el HANDSHAKE correctamente");
									send(auxPFS->cliente_pfs,&paquete_mando,sizeof(nipc),0);
								}
								else
								{
									paquete_mando.type = HANDSHAKE; //HANDSHAKE INCORRECTO
									paquete_mando.payloadlength = sizeof(nipc);
									strcpy(paquete_mando.payload.info,"HANDSHAKE INCORRECTO");
									if (configuracion.LogActivado==1)
										log_info(logger, "PPD - THREAD LECTOR PEDIDOS PFS", "Se recibio un HANDSHAKE incorrecto");
									send(auxPFS->cliente_pfs,&paquete_mando,sizeof(nipc),0);
								}
								break;
							default:
								//EL PEDIDO QUE LLEGO NO ES DE TIPO HANDSHAKE
								switch (algoritmoUtilizado)
								{
									case CON_INANICION:
										estructura = ppd_LBAToCHS(paquete_recibo.payload.direccionSector);
										if (configuracion.LogActivado==1)
											log_info(logger, "PPD - THREAD LECTOR PEDIDOS PFS", "Se agrego un pedido de %c sobre el sector %d:%d a la cola \n", paquete_recibo.payload.lectura_o_escritura, estructura.cylinder, estructura.sector);
										pthread_mutex_lock (&mutexCola);
										agregarPedido_cola(paquete_recibo.payload);
										pthread_mutex_unlock (&mutexCola);
										break;
									case SIN_INANICION:
										estructura = ppd_LBAToCHS(paquete_recibo.payload.direccionSector);
										if (configuracion.LogActivado==1)
											log_info(logger, "PPD - THREAD LECTOR PEDIDOS PFS", "Se agrego un pedido de %c sobre el sector %d:%d a la cola \n", paquete_recibo.payload.lectura_o_escritura, estructura.cylinder, estructura.sector);
										pthread_mutex_lock (&mutexCola);
										agregarPedido_cola(paquete_recibo.payload);
										pthread_mutex_unlock (&mutexCola);
										break;
								}
								break;
						}
						break;
				}
			}
			auxPFS=auxPFS->sig;
		}
	}
}

void insertarEnListaClientes(int32_t cliente_pfs){
	NodoPFS *aux;

	//CREO EL NODO A INSERTAR EN LA LISTA
	NodoPFS *nuevo=(NodoPFS*)malloc(sizeof(NodoPFS));

	nuevo->sig=NULL;
	nuevo->cliente_pfs=cliente_pfs;

	if( listaPFS == NULL )
	{
		listaPFS = nuevo;
	}
	else
	{
		aux = listaPFS;
		while( aux->sig != NULL )
		{
			aux = aux->sig;
		}

		aux->sig = nuevo;
	}

	cantidadClientesPFS++;
}

void borrarDeListaClientes(int32_t cliente_pfs){

	NodoPFS *aux, *anterior;

	anterior = NULL;

	for (aux = listaPFS;
			aux != NULL;
			anterior = aux, aux = aux->sig) {


		if (((aux->cliente_pfs == cliente_pfs))) {  /* Encontrado */
			if (anterior == NULL) {

				listaPFS = aux->sig;
			} else {

				anterior->sig = aux->sig;
			}

			free(aux);

			return;
		}
	}
	cantidadClientesPFS--;

}


